// -*-mode:c++; coding:utf-8-*-

#include "logger.hpp"

#include "multi_row_mutator_on_tablet.hpp"

namespace xbase
{
	typedef multi_row_mutator_on_tablet::handler_type handler_type;
	typedef multi_row_mutator_on_tablet::container_type container_type;
	typedef multi_row_mutator_on_tablet::container_iter container_iter;

	typedef multi_row_mutator_on_tablet::batch_write_req_ptr batch_write_req_ptr;

	// 在 lower_level_service 上执行，除第一次异步提交外
	// （为优化，直接由上层调用，而不再转到自己的
	// lower_level_service 上调用）
	//
	// 否则，在 async_apply 中应该是这样调用 on_async_apply
	// this->get_lower_level_service().dispatch(boost::bind(&this_type::on_async_apply,
	// 						        this_ptr()));
	void multi_row_mutator_on_tablet::on_async_apply()
	{
		m_session->async_write(m_req_ptr,
				       this->get_lower_level_service().wrap(
					       boost::bind(&this_type::on_response,
							   this_ptr(),
							   _1, _2)));
	}

	// 以下函数总在 lower_level_service 上执行

	void multi_row_mutator_on_tablet::on_response(protocol::packet_ptr ptr,
						      const error_code &socket_ec)
	{
		if(socket_ec) // 这是收发包出错，需要换session重试，由上层处理
		{
			on_error(0, socket_ec);
		}
		else if(ptr->get_type() != batch_write_resp::type)
		{
			on_error(0, error::bad_packet_type);
		}
		else
		{
			const batch_write_resp * resp = boost::polymorphic_downcast<batch_write_resp*>(ptr.get());
			const protocol::batch_write_response_v2 &body = resp->body();
			error_code ec = body.error_code();
			if(! ec) // 全部成功
			{
				dispatch(body.m_success_row,
					 ec);
			}
			else
			{
				if(ec == error::tablet_closed)
				{
					on_retry(body.m_success_row, ec);
				}
				else
				{
					on_error(body.m_success_row, ec);
				}
			}
		}
	}

	void multi_row_mutator_on_tablet::on_retry(std::size_t success_row, const error_code &ec)
	{
		if(m_options.can_retry(m_retry_times))
		{
			LOG(WARN, "mutator fail: " << ec << " - " << ec.message()
			    << ", retry_times = " << m_retry_times
			    << ", already success_row = " << m_success_row
			    << ", cur success_row = " << success_row
			    << ", cur total row = " << m_req_ptr->body().m_rows.size());

			m_req_ptr->body().remove_success_row(success_row);
			m_success_row += success_row;
			++ m_retry_times;

			if(! m_timer) // 延迟创建，因为大多数情况下不需要重试
				m_timer.reset(new boost::asio::deadline_timer(this->get_lower_level_service()));
			m_timer->expires_from_now(m_options.get_retry_interval());
			m_timer->async_wait(boost::bind(&this_type::on_async_apply,
							this_ptr()));
		}
		else
		{
			on_error(success_row, ec);
		}
	}

	void multi_row_mutator_on_tablet::on_error(std::size_t success_row, const error_code &ec)
	{
		LOG(ERROR, "mutator error: " << ec << " - " << ec.message()
		    << ", retry_times = " << m_retry_times
		    << ", already success_row = " << m_success_row
		    << ", cur success_row = " << success_row
		    /*<< ", request = "<< m_req_ptr->to_string()*/);
		dispatch(success_row, ec);
	}

	void multi_row_mutator_on_tablet::dispatch(std::size_t success_row, const error_code &ec)
	{
		const std::size_t total_success_row = m_success_row + success_row;

		// release resource before dispatch
		m_session.reset();
		m_req_ptr.reset();
		// timer在下次提交时还会用到，暂不释放

		handler_type h = m_handler;
		m_handler = handler_type(); // clear handler

		assert(h);
		this->get_upper_level_service().dispatch(
			boost::bind(h,
				    total_success_row,
				    ec)
			);
	}
}
